#include <sys/epoll.h>
#include <sys/eventfd.h>
#include "tcp_server.h"

namespace mod {

static const string MODULE = "TCPSVR";

Listener::Listener()
{
    listen_fd = -1;
    epoll_fd = -1;
    event_fd = -1;

    alive = false;
    thread = nullptr;
    signal(SIGPIPE, SIG_IGN);
}

Listener::~Listener()
{
    stop();
}

int Listener::start(const Config &conf)
{
	sockaddr_t sa;

    if (conf.local_af == AF_INET)
    {
        struct in_addr in;
        if (inet_pton(AF_INET, conf.local_addr.c_str(), (void *)&in) != 1)
            return 1;

        struct sockaddr_in *sa_in = (struct sockaddr_in *)&sa;
        sa_in->sin_family = AF_INET;
        sa_in->sin_port = htons((u16)conf.local_port);
        sa_in->sin_addr = in;

        listen_fd = ::socket(AF_INET, SOCK_STREAM, 0);
    }
    else if (conf.local_af == AF_INET6)
    {
        struct in6_addr in;
        if (inet_pton(AF_INET6, conf.local_addr.c_str(), (void *)&in) != 1)
            return 1;

        struct sockaddr_in6 *sa_in = (struct sockaddr_in6 *)&sa;
        sa_in->sin6_family = AF_INET6;
        sa_in->sin6_port = htons((u16)conf.local_port);
        sa_in->sin6_addr = in;

        listen_fd = ::socket(AF_INET6, SOCK_STREAM, 0);
    }
    else
    {
        logger.log_error(MODULE, "local addr is illegal");
        return 2;
    }

	if (listen_fd < 0)
	{
	    logger.log_error(MODULE, "fail to alloc listen fd");
	    return 3;
	}

	int ctl = fcntl(listen_fd, F_GETFL);
	fcntl(listen_fd, F_SETFL, ctl | O_NONBLOCK);

	int opt = 1;
	setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, (const char *)&opt, sizeof(opt));

	if (bind(listen_fd, &sa, sizeof(sa)) != 0)
	{
	    logger.log_error(MODULE, "fail to bind listen addr");
	    closesocket(listen_fd);
	    return 4;
	}

	event_fd = eventfd(0, EFD_NONBLOCK);
    if (event_fd < 0)
    {
        logger.log_error(MODULE, "fail to alloc event fd");
        closesocket(listen_fd);
        return 5;
    }

	epoll_fd = epoll_create1(0);
	if (epoll_fd < 0)
	{
	    logger.log_error(MODULE, "fail to alloc epoll fd");
        closesocket(event_fd);
	    closesocket(listen_fd);
	    return 5;
	}

    //select listen fd
    {
        listen_dock.cid = -1;
        listen_dock.sock = listen_fd;

    	struct epoll_event ee;
    	ee.events = EPOLLIN | EPOLLET;
    	ee.data.ptr = &listen_dock;
    	epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ee);
    }

    //select event fd
    {
        event_dock.cid = -1;
        event_dock.sock = event_fd;

    	struct epoll_event ee;
    	ee.events = EPOLLIN | EPOLLET;
    	ee.data.ptr = &event_dock;
    	epoll_ctl(epoll_fd, EPOLL_CTL_ADD, event_fd, &ee);
    }

    config = conf;
	alive = true;
	thread = new std::thread(entry, this);
    return 0;
}

void Listener::stop()
{
    if (alive)
    {
        alive = false;
        thread->join();
	    closesocket(epoll_fd);
	    closesocket(event_fd);
        closesocket(listen_fd);
    }
}

void Listener::send(u64 cid, const string &str)
{
    Event evt;
    evt.cid = cid;
    evt.str = str;
    cache_queue.push(evt);

    uint64_t num = 1;
    write(event_fd, &num, sizeof(num));
}

void Listener::send(u64 cid, string &&str)
{
    Event evt;
    evt.cid = cid;
    evt.str = std::move(str);
    cache_queue.push(evt);

    uint64_t num = 1;
    write(event_fd, &num, sizeof(num));
}

void Listener::close(u64 cid)
{
    Event evt;
    evt.cid = cid;
    evt.buf = NULL;
    evt.len = 0;
    cache_queue.push(evt);

    uint64_t num = 1;
    write(event_fd, &num, sizeof(num));
}

void Listener::epoll_add(Dock *dock)
{
    struct epoll_event ee;
    ee.events = EPOLLIN | EPOLLET;
    ee.data.ptr = dock;
    epoll_ctl(epoll_fd, EPOLL_CTL_ADD, dock->sock, &ee);
}

void Listener::epoll_del(Dock *dock)
{
    struct epoll_event ee;
    epoll_ctl(epoll_fd, EPOLL_CTL_DEL, dock->sock, &ee);
}

void Listener::epoll_ro(Dock *dock)
{
    struct epoll_event ee;
    ee.events = EPOLLIN | EPOLLET;
    ee.data.ptr = dock;
    epoll_ctl(epoll_fd, EPOLL_CTL_MOD, dock->sock, &ee);
}

void Listener::epoll_rw(Dock *dock)
{
    struct epoll_event ee;
    ee.events = EPOLLIN | EPOLLOUT |EPOLLET;
    ee.data.ptr = dock;
    epoll_ctl(epoll_fd, EPOLL_CTL_MOD, dock->sock, &ee);
}

Listener::Dock *Listener::alloc_dock(int fd)
{
    Listener::Dock *dock = dock_pool.alloc();
    if (!dock)
        return nullptr;

    dock->cid = util::get_unique_id();
    dock->sock = fd;
    dock->buffer = NULL;
    dock->offset = 0;
    dock->length = 0;

    dock_map.insert(std::make_pair(dock->cid, dock));
    epoll_add(dock);

    logger.log_debug(MODULE, "dock[%llx] is allocated", dock->cid);
    return dock;
}

void Listener::free_dock(Listener::Dock *dock)
{
    logger.log_debug(MODULE, "dock[%llx] is freed", dock->cid);

    epoll_del(dock);
    dock_map.erase(dock->cid);

    dock->cid = 0;
    closesocket(dock->sock);
    dock->sock = -1;

    if (dock->buffer)
        free(dock->buffer);
    dock->buffer = NULL;
    dock->offset = 0;
    dock->length = 0;

    dock_pool.free(dock);
}

void Listener::period_handler()
{

}

void Listener::event_handler(Dock *dock, struct epoll_event &ee)
{
    if (dock->sock == listen_fd)
    {
        while (true)
        {
            sockaddr_t addr;
            socklen_t addr_len = sizeof(addr);

            int connect_fd = accept(listen_fd, &addr, &addr_len);
            if(connect_fd < 0)
                break;

            int ctl = fcntl(connect_fd, F_GETFL);
            fcntl(connect_fd, F_SETFL, ctl | O_NONBLOCK);

            Dock *new_dock = alloc_dock(connect_fd);
            if (new_dock == nullptr)
            {
                logger.log_notice(MODULE, "fail to alloc dock");
                closesocket(connect_fd);
                break;
            }

            char remote_addr[64];
            int remote_port;

            if (config.local_af == AF_INET)
            {
                struct sockaddr_in *sa_in = (struct sockaddr_in *)&addr;
                inet_ntop(AF_INET, &sa_in->sin_addr, new_dock->addr, sizeof(new_dock->addr));
                new_dock->port = ntohs(sa_in->sin_port);
            }
            else
            {
                assert(config.local_af == AF_INET6);

                struct sockaddr_in6 *sa_in6 = (struct sockaddr_in6 *)&addr;
                inet_ntop(AF_INET6, &sa_in6->sin_addr, new_dock->addr, sizeof(new_dock->addr));
                new_dock->port = ntohs(sa_in6->sin_port);
            }

            config.accepted(new_dock->cid, new_dock->addr, new_dock->port);
        }
    }
    else if (dock->sock == event_fd)
    {
        int total = cache_queue.size();
        Event evt;

        for (int i=0; i<total; ++i)
        {
            cache_queue.pop(evt);

            Dock *got_dock = dock_map[evt.cid];
            if (!got_dock)
            {
                logger.log_notice(MODULE, "cid not found %llx", evt.cid);
                continue;
            }

            if (got_dock->cid != evt.cid)
            {
                logger.log_notice(MODULE, "cid not match %llx != %llx", got_dock->cid, evt.cid);
                continue;
            }

            if (evt.buf == NULL)
            {
                logger.log_debug(MODULE, "dock[%llx] is closing", got_dock->cid);
                free_dock(got_dock);
                continue;
            }
            else
            {
                char *tmp = (char *)realloc(got_dock->buffer, got_dock->length + evt.len);

                if (tmp == NULL)
                {
                    logger.log_notice(MODULE, "dock[%llx] realloc return null", got_dock->cid);
                    free(evt.buf);
                    continue;
                }

                memcpy(tmp + got_dock->length, evt.buf, evt.len);
                got_dock->buffer = tmp;
                got_dock->length = got_dock->length + evt.len;

                epoll_rw(got_dock);
            }
        }
    }
    else if(ee.events & EPOLLIN)
    {
        while (true)
        {
            char *buf = new char[2048];

            int len = read(dock->sock, buf, 2048);
            if (len <= 0)
            {
                free(buf);

                if (errno == EAGAIN)
                    break;

                logger.log_info(MODULE, "dock[%llx] connection is closed", dock->cid);
                config.closed(dock->cid);
                free_dock(dock);
                break;
            }

            logger.log_debug(MODULE, "dock[%llx] receive %d bytes data", dock->cid, len);
            config.received(dock->cid, dock->addr, dock->port, strref(std::move(buf), len));
        }
    }
    else if(ee.events & EPOLLOUT)
    {
        if (dock->length <= 0)
        {
            epoll_ro(dock);
            return;
        }

        int len = write(dock->sock, dock->buffer+dock->offset, dock->length-dock->offset);
        if (len < 0)
        {
            logger.log_debug(MODULE, "dock[%llx] connection is closed", dock->cid);
            config.closed(dock->cid);
            free_dock(dock);
            return;
        }

        logger.log_debug(MODULE, "dock[%llx] sent out %d bytes data", dock->cid, len);

        dock->offset += len;

        if (dock->offset >= dock->length)
        {
            free(dock->buffer);
            dock->buffer = NULL;
            dock->offset = 0;
            dock->length = 0;

            epoll_ro(dock);
        }
    }
}

void Listener::main_entry()
{
	logger.log_notice(MODULE, "start listening connection");
	listen(listen_fd, SOMAXCONN);

    time_t check_point = time(NULL)+1;

	while (alive)
	{
		struct epoll_event ees[100];
		int nfds = epoll_wait(epoll_fd, ees, 100, 500);

		for (int i=0; i<nfds; ++i)
		{
		    Dock *dock = (Dock *)ees[i].data.ptr;
            event_handler(dock, ees[i]);
		}

        if (check_point < time(NULL))
        {
            period_handler();
            check_point = time(NULL) + 10;
        }
	}
}

void Listener::entry(Listener *tcpsvr)
{
    tcpsvr->main_entry();
}

}
